package com.example.flinkcourse.lesson1.sink;

import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * PostgreSQL Sink 工厂类
 * 用于创建 PostgreSQL 输出
 */
public class PgSinkFactory {
    private static final Logger LOG = LoggerFactory.getLogger(PgSinkFactory.class);

    /**
     * 创建 PostgreSQL Sink
     * @return JdbcSink<String>
     */
    public static JdbcSink<String> createPgSink() {
        LOG.info("Creating PostgreSQL sink...");

        // 配置 JDBC 连接
        JdbcConnectionOptions connectionOptions = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
            .withUrl("jdbc:postgresql://localhost:5432/flink_db")
            .withDriverName("org.postgresql.Driver")
            .withUsername("postgres")
            .withPassword("postgres")
            .build();

        // 配置执行选项
        JdbcExecutionOptions executionOptions = JdbcExecutionOptions.builder()
            .withBatchSize(1000)
            .withBatchIntervalMs(5000)
            .withMaxRetries(3)
            .build();

        // 创建 PostgreSQL Sink
        JdbcSink<String> sink = JdbcSink.sink(
            "INSERT INTO output_table (data, timestamp) VALUES (?, ?)",
            new JdbcStatementBuilder<String>() {
                @Override
                public void accept(PreparedStatement statement, String value) throws SQLException {
                    statement.setString(1, value);
                    statement.setTimestamp(2, new java.sql.Timestamp(System.currentTimeMillis()));
                }
            },
            executionOptions,
            connectionOptions
        );

        LOG.info("PostgreSQL sink created successfully");
        return sink;
    }
} 